import org.apache.spark.sql.SparkSession
import traffic_detect_monitorspeedrank.monitorSpeedRank.mysortBy

import scala.reflect.ClassTag

object traffic_detect_monitorspeedrank {


  object monitorSpeedRank{
    def speedCollection(tuple:((Any, Iterable[Any]))) = {
      val monitorId = tuple._1
      val speedCollection = tuple._2.iterator
      var highCount = 0
      var mediumCount = 0
      var lowCount = 0
      while (speedCollection.hasNext){
        var speed = speedCollection.next.asInstanceOf[String].toInt
        if (speed<100&&speed>0)
          lowCount+=1
        else if (speed<=200&&speed>=100)
          mediumCount+=1
        else
          highCount+=1
      }
      (monitorId, (highCount.toString, mediumCount.toString, lowCount.toString))
    }

    val mysortBy = new Ordering[Tuple3[String, String, String]] {
      override def compare(x: (String, String, String), y: (String, String, String)): Int = {
        val r = x._1.compare(y._1)
        val r2 = x._2.compare(y._2)
        if (r == 0) {
          if (r2 == 0) y._3.compare(x._3)  else r2
        } else r
      }
    }




}

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("traffic_detection").getOrCreate()
    spark.sql("use traffic")
    val flow_rdd = spark.sql("select * from monitor_flow_action").rdd
    flow_rdd.map(row=>{(row.get(1),row.get(5))}).groupByKey().map(monitorSpeedRank.speedCollection).sortBy(x=>x._2)(mysortBy, ClassTag.apply[Tuple3[String, String, String]](classOf[Tuple3[String, String, String]])).collect().foreach(println)
  }



}
